package com.education.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.education.realtime.app.BaseAppV2;
import com.education.realtime.bean.TrafficCountBean;
import com.education.realtime.util.AtguiguUtil;
import com.education.realtime.util.FlinkSinkUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.time.Duration;
import java.util.Map;

import static com.education.realtime.common.Constant.*;

public class DwsTrafficPageView extends BaseAppV2 {
    public static void main(String[] args) {
        new DwsTrafficPageView().init(3001,"DwsTrafficPageView",
                TOPIC_DWD_TRAFFIC_PAGE,
                TOPIC_DWD_TRAFFIC_UV_DETAIL,
                TOPIC_DWD_TRAFFIC_UJ_DETAIL);
    }

    @Override
    protected void handle(StreamExecutionEnvironment env, Map<String, DataStreamSource<String>> stream) {
        // 1. 多个流union成一个流
        DataStream<TrafficCountBean> trafficPageViewBeanStream = unionToOne(stream);
        // 2. 分组开窗聚合
        SingleOutputStreamOperator<TrafficCountBean> resultStream = windowAndAggregate(trafficPageViewBeanStream);
        // 3. 结果写入到clickhouse中
        writeToClickHouse(resultStream);
    }

    private void writeToClickHouse(SingleOutputStreamOperator<TrafficCountBean> resultStream) {
        resultStream.addSink(FlinkSinkUtil.getClickHouseSink(CLICKHOUSE_DB,
                "dws_traffic_page_view",
                TrafficCountBean.class
        ));
    }

    private SingleOutputStreamOperator<TrafficCountBean> windowAndAggregate(DataStream<TrafficCountBean> stream) {
        /*
        开窗: 用什么时间? 事件时间
          滚动窗口

          reduce
         */
        return stream
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy
                                .<TrafficCountBean>forBoundedOutOfOrderness(Duration.ofSeconds(20))
                                .withTimestampAssigner((bean, ts) -> bean.getTs())
                )
                .keyBy(bean ->  bean.getCh())
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                .sideOutputLateData(new OutputTag<TrafficCountBean>("late") {})
                .reduce(
                        new ReduceFunction<TrafficCountBean>() {
                            @Override
                            public TrafficCountBean reduce(TrafficCountBean b1,
                                                              TrafficCountBean b2) throws Exception {
                                b1.setPageCount(b1.getPageCount()+b2.getPageCount());
                                b1.setSessionCount(b1.getSessionCount()+b2.getSessionCount());
                                b1.setTimeCount(b1.getTimeCount()+b2.getTimeCount());
                                b1.setUv(b1.getUv()+b2.getUv());
                                b1.setUj(b1.getUj()+b2.getUj());

                                return b1;
                            }
                        },
                        new ProcessWindowFunction<TrafficCountBean, TrafficCountBean, String, TimeWindow>() {
                            @Override
                            public void process(String key,
                                                Context ctx,
                                                Iterable<TrafficCountBean> elements,
                                                Collector<TrafficCountBean> out) throws Exception {
                                TrafficCountBean bean = elements.iterator().next();
                                String stt = AtguiguUtil.toDateTime(ctx.window().getStart());
                                String edt = AtguiguUtil.toDateTime(ctx.window().getEnd());

                                bean.setStt(stt);
                                bean.setEdt(edt);

                                bean.setTs(System.currentTimeMillis());  //把ts设置为系统时间, 表示这个结果的统计时间

                                out.collect(bean);

                            }
                        }
                );

    }

    private DataStream<TrafficCountBean> unionToOne(Map<String, DataStreamSource<String>> topicAndStreamMap) {
        SingleOutputStreamOperator<TrafficCountBean> pvSvDuringTimeStream = topicAndStreamMap
                .get(TOPIC_DWD_TRAFFIC_PAGE)
                .map(new MapFunction<String, TrafficCountBean>() {
                    @Override
                    public TrafficCountBean map(String value) throws Exception {
                        JSONObject obj = JSON.parseObject(value);
                        JSONObject common = obj.getJSONObject("common");
                        JSONObject page = obj.getJSONObject("page");

                        String ch = common.getString("ch");
                        Long ts = obj.getLong("ts");

                        Long uv = 0l;
                        Long uj = 0l;
                        Long pageCount=1l;
                        Long sessionCount=page.getLong("last_page_id")== null ? 1l : 0l;
                        Long timeCount=page.getLong("during_time");

                        return new TrafficCountBean("", "",
                                ch,ts,uv,uj,pageCount,sessionCount,timeCount
                        );

                    }
                });

        SingleOutputStreamOperator<TrafficCountBean> uvStream = topicAndStreamMap
                .get(TOPIC_DWD_TRAFFIC_UV_DETAIL)
                .map(json -> {

                    JSONObject obj = JSON.parseObject(json);
                    JSONObject common = obj.getJSONObject("common");
                    JSONObject page = obj.getJSONObject("page");

                    String ch = common.getString("ch");
                    Long ts = obj.getLong("ts");

                    Long uv = 1l;
                    Long uj = 0l;
                    Long pageCount=0l;
                    Long sessionCount=0l;
                    Long timeCount=0l;

                    return new TrafficCountBean("", "",
                            ch,ts,uv,uj,pageCount,sessionCount,timeCount
                    );


                });


        SingleOutputStreamOperator<TrafficCountBean> ujStream = topicAndStreamMap
                .get(TOPIC_DWD_TRAFFIC_UJ_DETAIL)
                .map(json -> {
                    JSONObject obj = JSON.parseObject(json);
                    JSONObject common = obj.getJSONObject("common");
                    JSONObject page = obj.getJSONObject("page");

                    String ch = common.getString("ch");
                    Long ts = obj.getLong("ts");

                    Long uv = 0l;
                    Long uj = 1l;
                    Long pageCount=0l;
                    Long sessionCount=0l;
                    Long timeCount=0l;

                    return new TrafficCountBean("", "",
                            ch,ts,uv,uj,pageCount,sessionCount,timeCount
                    );

                });

        return pvSvDuringTimeStream.union(ujStream, uvStream);
    }
}
